package com.lhd.app.func;


import com.alibaba.druid.pool.DruidDataSource;
import com.alibaba.druid.pool.DruidPooledConnection;
import com.alibaba.fastjson.JSONObject;
import com.lhd.common.utils.DimUtil;
import com.lhd.common.utils.DruidDSUtil;
import com.lhd.common.utils.ThreadPoolUtil;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.async.ResultFuture;
import org.apache.flink.streaming.api.functions.async.RichAsyncFunction;

import java.util.Collections;
import java.util.concurrent.ThreadPoolExecutor;

/*

流：  对象    id   拿着id到habse里面 去查（动态）

 */
public abstract class DimAsyncFunction<T> extends RichAsyncFunction<T, T> implements DimJoinFunction<T> {




    private DruidDataSource dataSource;
    private ThreadPoolExecutor threadPoolExecutor;

    private String tableName;

    public DimAsyncFunction() {
    }

    public DimAsyncFunction(String tableName) {
        this.tableName = tableName;
    }
//    abstract String getKey(T input);
//
//    abstract  void join(T input, JSONObject dimInfo);
    //连接池
    @Override
    public void open(Configuration parameters) throws Exception {
        dataSource = DruidDSUtil.createDataSource();
        threadPoolExecutor = ThreadPoolUtil.getThreadPoolExecutor();
    }



    //这个被加载的时候会触发执行asyncInvoke
    /*
     线程池   4个核心线程  20非核心线程     100秒如果线程不操作就会被回收
    * 1.开启线程
    * 2.执行任务
      3. 任务是获取流里面的数据  根据id关联查询 spu   tm   c3
      4.最终把我们的数据输出
    * */
    //sku_info  id       tm_id  spu_id
    //tm_id    tmname
    //spu_id   spuname
    @Override
    public void asyncInvoke(T input, ResultFuture<T> resultFuture) throws Exception {
       //异步IO
        threadPoolExecutor.execute(new Runnable() {

            //sku_info      获取连接     书写sql    给对象赋值  把结果返回    变成活的
            //SKU_INFO       getSkuID
            //TM_INFO        getTmId  活的
//            sql 旁路缓存    jdbcutils.queryList()
            @Override
            public void run() {

                //连接habse  把结果出来玩  封装对象里面进行返回
                try {
                    //获取连接
                    DruidPooledConnection connection = dataSource.getConnection();
                    //查询维表获取维度信息  input {}
                    String key = getKey(input);//sku_id   user_id  g3_id
                    //表名用成活的通过构造器
//                    JdbcUtil.
                    //key  使用模板方法  说白了就是接口 实现接口但是不实现方法
                    JSONObject dimInfo = DimUtil.getDimInfo(connection, tableName, key);
//                 T      spu_id    tm_id   a3_id
//                 T      spu_name
//                 T      tm_name
//                 T      a3_name
                    //将维度信息补充至当前数据
                    if (dimInfo != null) {
                        join(input, dimInfo);
                    }

                    //归还连接
                    connection.close();

                    //将结果写出
                    resultFuture.complete(Collections.singletonList(input));


                } catch (Exception e) {
                    e.printStackTrace();
                    System.out.println("关联维表失败：" + input + ",Table:" + tableName);
                    //resultFuture.complete(Collections.singletonList(input));
                }
            }
        });
    }

    //在指定的时间内没有完成此任务  打印输出

    @Override
    public void timeout(T input, ResultFuture<T> resultFuture) throws Exception {
        System.out.println("TimeOut:" + input);
    }


}
